package com.qqt.csr.im.service.mongo;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.mongodb.client.result.UpdateResult;
import com.qqt.csr.common.exception.BaseException;
import com.qqt.csr.common.exception.StatusCode;
import com.qqt.csr.common.utils.CacheKeyUtil;
import com.qqt.csr.common.utils.RedissonLockUtil;
import com.qqt.csr.common.utils.SnowflakeUtil;
import com.qqt.csr.im.assembler.ConversationAssembler;
import com.qqt.csr.im.assembler.RoomAssembler;
import com.qqt.csr.im.dto.BindResultDTO;
import com.qqt.csr.im.dto.ConversationDTO;
import com.qqt.csr.im.dto.req.CreatingConversationReqDTO;
import com.qqt.csr.im.dto.resp.CreatingConversationRespDTO;
import com.qqt.csr.im.entity.*;
import com.qqt.csr.im.entity.mq.MqMessage;
import com.qqt.csr.im.enums.ConversationStateEnum;
import com.qqt.csr.im.mapper.mongo.ConversationMapper;
import com.qqt.csr.im.service.*;
import com.qqt.csr.im.socketio.constants.EventsConstants;
import com.qqt.csr.im.utils.DeviceCacheKeyUtil;
import com.qqt.csr.im.vo.resp.RoomRespVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @Author: yang hao
 * @Date: 2024/10/15 9:50
 * @Description: 会话 业务层
 **/
@Slf4j
@Service
public class ConversationService {
    @Autowired
    private ConversationMapper conversationMapper;
    @Autowired
    private MongoTemplate mongoTemplate;
    @Autowired
    private AccountService accountService;
    @Autowired
    private ConnectionService connectionService;
    @Autowired
    private EventMsgProducerService producerService;
    @Autowired
    private RoomService roomService;
    @Autowired
    private RoomMemberService roomMemberService;
    @Autowired
    private RedissonLockUtil redissonLockUtil;

    /**
     * 获取最近一条未结束的会话
     *
     * @param initiatorId {@link String} 创建人账户id
     * @return {@link Conversation} 会话
     */
    public Conversation getLastNormalConversationBy(String initiatorId) {
        Criteria criteria = Criteria.where("initiatorId").is(initiatorId)
                .and("state").is(ConversationStateEnum.NORMAL.name());
        Sort sort = Sort.by(Sort.Direction.DESC, "createTime");
        Query query = Query.query(criteria).with(sort).limit(1);
        List<Conversation> conversations = mongoTemplate.find(query, Conversation.class);
        if (CollectionUtils.isEmpty(conversations)) {
            return null;
        }
        return conversations.get(0);
    }

    /**
     * 获取最近一条未结束的会话
     *
     * @param initiatorId {@link String} 创建人账户id
     * @return {@link Conversation} 会话
     */
    public Conversation getLastConversationBy(String initiatorId, String startTime, String endTime) {
        Criteria criteria = Criteria.where("initiatorId").is(initiatorId)
                .and("createTime").gte(startTime)
                .and("createTime").lte(endTime);
        Sort sort = Sort.by(Sort.Direction.DESC, "createTime");
        Query query = Query.query(criteria).with(sort).limit(1);
        List<Conversation> conversations = mongoTemplate.find(query, Conversation.class);
        if (CollectionUtils.isEmpty(conversations)) {
            return null;
        }
        return conversations.get(0);
    }

    /**
     * 创建会话
     *
     * @param dto
     * @return
     */
    public CreatingConversationRespDTO create(CreatingConversationReqDTO dto) {
        String lockKey = String.format(CacheKeyUtil.IM.CREATE_CONVERSATION_LOCK, dto.getAccountId().toString());
        return redissonLockUtil.distributedLock(() -> {
            Conversation conversation = ConversationAssembler.convert(dto.getAccountId().toString(), dto.getConversationExt());

            Conversation existConversation = getLastNormalConversationBy(dto.getAccountId().toString());
            if (existConversation != null) {
                conversation.setId(existConversation.getId());
            }
            conversationMapper.save(conversation);
            return CreatingConversationRespDTO.builder().conversationId(conversation.getId()).build();
        }, lockKey, 30, 1, TimeUnit.MINUTES);
    }

    /**
     * 房间绑定会话
     *
     * @param conversation
     * @param accountIdList
     * @return
     */
    public BindResultDTO bindRoom(Conversation conversation, List<Long> accountIdList) {
        if (CollectionUtils.isEmpty(accountIdList)) {
            return null;
        }

        String lockKey = String.format(CacheKeyUtil.IM.JOIN_CONVERSATION_LOCK, conversation.getId());
        return redissonLockUtil.distributedLock(() -> {
            Room room = roomService.queryRoomBy(accountIdList);
            if (room != null) {
                conversation.setRoomId(room.getId().toString());
                room.setExt(conversation.getExt());
                if (ConversationStateEnum.NORMAL.getCode().equals(room.getLastConversationState())) {
                    // 当前房间存在未结束会话
                    conversation.setId(room.getLastConversationId());
                    conversation.setStartTime(DateUtil.format(room.getLastConversationDate(), DatePattern.NORM_DATETIME_PATTERN));
                }
                conversationMapper.save(conversation);
                roomService.updateLastConversation(conversation, room);
            } else {
                room = new Room();
                room.setId(SnowflakeUtil.nextId());
                room.setExt(conversation.getExt());
                conversation.setRoomId(room.getId().toString());
                conversationMapper.save(conversation);
                roomService.save(accountIdList, conversation, room);
            }
            // 响应vo转换
            return ConversationAssembler.convert(conversation, room);
        }, lockKey, 30, 1, TimeUnit.MINUTES);

    }


    // /**
    //  * 会话创建
    //  *
    //  * @param dto {@link BindDTO} 绑定参数
    //  * @return {@link BindResultDTO} 绑定结果
    //  */
    // public BindResultDTO create(BindDTO dto) {
    //     Account initiator = dto.getInitiator();
    //     Account recipient = dto.getRecipient();
    //     if (initiator == null || recipient == null) {
    //         throw new BaseException(StatusCode.Common.ILLEGAL_PARAM_ERROR);
    //     }
    //
    //     String lockKey = String.format(CacheKeyUtil.IM.CREATE_CONVERSATION_LOCK, initiator.getId(), recipient.getId());
    //     return redissonLockUtil.distributedLock(() -> {
    //         // 实体转换
    //         Conversation conversation = ConversationAssembler.convert(initiator.getId().toString());
    //         conversation.setExt(dto.getConversationExt());
    //
    //         Room room = roomService.queryRoomBy(Arrays.asList(initiator.getId(), recipient.getId()));
    //         if (room != null) {
    //             conversation.setRoomId(room.getId().toString());
    //             room.setExt(dto.getRoomExt());
    //             if (ConversationStateEnum.NORMAL.getCode().equals(room.getLastConversationState())) {
    //                 // 当前房间存在未结束会话
    //                 conversation.setId(room.getLastConversationId());
    //                 conversation.setStartTime(DateUtil.format(room.getLastConversationDate(), DatePattern.NORM_DATETIME_PATTERN));
    //             }
    //             conversationMapper.save(conversation);
    //             roomService.updateLastConversation(conversation, room);
    //         } else {
    //             room = new Room();
    //             room.setId(SnowflakeUtil.nextId());
    //             room.setExt(dto.getRoomExt());
    //             conversation.setRoomId(room.getId().toString());
    //             conversationMapper.save(conversation);
    //             roomService.save(initiator, recipient, conversation, room);
    //         }
    //         // 响应vo转换
    //         return ConversationAssembler.convert(conversation, initiator, recipient, room);
    //     }, lockKey, 30, 1, TimeUnit.MINUTES);
    // }

    /**
     * 获取前5次会话列表
     *
     * @param conversationId {@link String} 会话id
     * @return {@link List<Conversation>} 会话列表
     */
    public List<Conversation> lastFiveConversation(String conversationId) {
        Conversation conversation = getById(conversationId);
        if (Objects.isNull(conversation)) {
            throw new BaseException(StatusCode.Common.ILLEGAL_PARAM_ERROR);
        }
        Criteria criteria = Criteria.where("roomId").is(conversation.getRoomId());
        criteria.and("createTime").lt(conversation.getCreateTime());
        Sort sort = Sort.by(Sort.Direction.DESC, "createTime");
        Query query = Query.query(criteria).with(sort).limit(5);
        return mongoTemplate.find(query, Conversation.class);
    }

    /**
     * 获取会话列表
     *
     * @param accountUserId im账户的用户标识
     * @param size          查询数量，如果传0，则无查询数量限制
     * @return
     */
    public List<ConversationDTO> queryConversationListBy(String accountUserId, ConversationStateEnum stateEnum, int size) {
        Account account = accountService.getByAccountUserId(accountUserId);
        if (account == null) {
            return Lists.newArrayList();
        }
        return queryConversationListBy(account, stateEnum, size);
    }

    /**
     * 获取会话列表
     *
     * @param account im账户的用户标识
     * @param size    查询数量，如果传0，则无查询数量限制
     * @return
     */
    public List<ConversationDTO> queryConversationListBy(Account account, ConversationStateEnum stateEnum, int size) {
        if (account == null) {
            return Lists.newArrayList();
        }

        List<String> roomIdStrList = roomService.queryRoomList(account.getId(), stateEnum)
                .stream().map(room -> room.getId().toString()).collect(Collectors.toList());
        if (CollectionUtils.isEmpty(roomIdStrList)) {
            return Lists.newArrayList();
        }

        Criteria criteria = null;
        if (roomIdStrList.size() == 1) {
            criteria = Criteria.where("roomId").is(roomIdStrList.get(0));
        } else {
            criteria = Criteria.where("roomId").in(roomIdStrList);
        }
        Query query = new Query();
        query.addCriteria(criteria.and("state").is(stateEnum.name()));
        query.with(Sort.by(Sort.Direction.DESC, "createTime"));
        if (size > 0) {
            query.limit(size);
        }
        List<Conversation> conversations = mongoTemplate.find(query, Conversation.class);

        List<Long> roomIdList = conversations.stream().map(Conversation::getRoomId).distinct().map(Long::parseLong).collect(Collectors.toList());
        Map<Long, List<RoomMember>> roomMemMap = roomMemberService.queryMemberMapFromCacheBy(roomIdList);

        List<Long> accountIdList = roomMemMap.values().stream().flatMap(Collection::stream)
                .map(RoomMember::getAccountId).distinct().filter(aId -> !aId.equals(account.getId())).collect(Collectors.toList());
        Map<Long, Account> accountMap = accountService.getMapByIds(accountIdList);
        accountMap.put(account.getId(), account);

        return ConversationAssembler.convert(conversations, accountMap, roomMemMap);
    }

    /**
     * 结束会话
     *
     * @param conversionId {@link String} 会话id
     * @return {@link Boolean} 是否结束
     */
    public void end(String conversionId) {
        Conversation conversation = getById(conversionId);
        String date = DateUtil.now();
        if (conversation != null) {
            conversation.setState(ConversationStateEnum.END);
            conversation.setEndTime(date);
            conversation.setUpdateTime(date);
            conversationMapper.save(conversation);
        }
        // 更新房间
        roomService.updateRoomCloseConversationState(Lists.newArrayList(conversionId));
    }

    /**
     * 结束会话
     *
     * @param conversionIdList {@link String} 会话id
     * @return {@link Boolean} 是否结束
     */
    public void end(List<String> conversionIdList) {
        if (CollectionUtils.isEmpty(conversionIdList)) {
            return;
        }

        String date = DateUtil.now();
        Criteria criteria = Criteria.where("_id").in(conversionIdList);
        Update update = Update.update("state", ConversationStateEnum.END)
                .set("endTime", date)
                .set("updateTime", date);
        UpdateResult updateResult = mongoTemplate.updateMulti(Query.query(criteria), update, Conversation.class);
        if (updateResult.getModifiedCount() > 0) {
            // 更新房间
            roomService.updateRoomCloseConversationState(conversionIdList);
        }
    }

    /**
     * 通过id获取会话信息
     *
     * @param conversationId {@link String} 会话id
     * @return {@link Conversation} 会话信息
     */
    public Conversation getById(String conversationId) {
        if (StringUtils.isBlank(conversationId)) {
            return null;
        }
        Optional<Conversation> result = conversationMapper.findById(conversationId);
        return result.orElse(null);
    }

    /**
     * 通过id获取会话信息
     *
     * @param conversationIdList {@link String} 会话id
     * @return {@link Conversation} 会话信息
     */
    public Map<String, Conversation> getById(List<String> conversationIdList) {
        if (CollectionUtils.isEmpty(conversationIdList)) {
            return Maps.newHashMap();
        }
        List<Conversation> list = conversationMapper.findAllById(conversationIdList);
        return list.stream().collect(Collectors.toMap(Conversation::getId, v -> v, (o, n) -> o));
    }

    /**
     * 会话拉取通知
     *
     * @param account {@link String} 用户账户
     * @return {@link Boolean} 是否通知成功
     */
    public void pushNewConversationEvent(Account account, Room room, Map<Long, Account> accountMap) {
        List<Connection> connectionList = connectionService.getOnlineByAccountId(account.getId());
        for (Connection connection : connectionList) {
            // 发送消息通知
            String deviceKey = DeviceCacheKeyUtil.getKey(connection);
            MqMessage data = MqMessage.of(connection.getServiceId(), deviceKey, null, RoomAssembler.convertBy(room, account, accountMap));
            producerService.sendEventMsg(EventsConstants.NEW_CONVERSATION_EVENT, data);
        }
    }

    /**
     * 推送关闭会话事件
     *
     * @param account
     * @param conversation
     */
    public void pushCloseConversationEvent(Account account, Conversation conversation) {
        if(account == null || conversation == null){
            return;
        }
        Map<String, List<Account>> accountNotifyMap = Maps.newHashMap();
        accountNotifyMap.put(conversation.getId(), Lists.newArrayList(account));
        pushCloseConversationEvent(accountNotifyMap, Lists.newArrayList(conversation));
    }

    /**
     * 推送关闭会话事件
     *
     * @param account
     * @param conversationList
     */
    public void pushCloseConversationEvent(Account account, List<ConversationDTO> conversationList) {
        if (CollectionUtils.isEmpty(conversationList)) {
            return;
        }

        List<Connection> connectionList = connectionService.getOnlineByAccountId(account.getId());
        for (ConversationDTO conversationDTO : conversationList) {
            RoomRespVO roomRespVO = RoomAssembler.convertBy(conversationDTO);
            roomRespVO.setLastConversationState(ConversationStateEnum.END.getCode());
            for (Connection connection : connectionList) {
                // 发送消息通知
                String deviceKey = DeviceCacheKeyUtil.getKey(connection);
                MqMessage data = MqMessage.of(connection.getServiceId(), deviceKey, null, roomRespVO);
                producerService.sendEventMsg(EventsConstants.CLOSED_CONVERSATION_EVENT, data);
            }
        }

    }

    /**
     * 推送关闭会话事件
     *
     * @param colsedConversationAccountNotifyMap key:会话id；value：需要接收通知的账户列表
     * @param closingConversationList
     */
    public void pushCloseConversationEvent(Map<String, List<Account>> colsedConversationAccountNotifyMap, List<Conversation> closingConversationList) {
        if (CollectionUtils.isEmpty(closingConversationList)) {
            return;
        }

        List<Long> accountIdList = colsedConversationAccountNotifyMap.values().stream()
                .flatMap(Collection::stream)
                .map(Account::getId).collect(Collectors.toList());
        Map<Long, List<Connection>> connectionMap = connectionService.getOnlineByAccountId(accountIdList);
        for (Conversation conversation : closingConversationList) {
            List<Account> accountList = colsedConversationAccountNotifyMap.get(conversation.getId());
            if (CollectionUtils.isEmpty(accountList)) {
                continue;
            }

            RoomRespVO roomRespVO = RoomAssembler.convertBy(conversation);
            roomRespVO.setLastConversationState(ConversationStateEnum.END.getCode());
            accountList.forEach(account -> {
                Optional.ofNullable(connectionMap.get(account.getId())).orElseGet(Lists::newArrayList)
                        .forEach(connection -> {
                            // 发送消息通知
                            String deviceKey = DeviceCacheKeyUtil.getKey(connection);
                            MqMessage data = MqMessage.of(connection.getServiceId(), deviceKey, null, roomRespVO);
                            producerService.sendEventMsg(EventsConstants.CLOSED_CONVERSATION_EVENT, data);
                        });
            });
        }
    }
}
